package com.jkgh.asin;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;

import com.jkgh.asin.buffers.ProducerConsumerBuffer;
import com.jkgh.asin.buffers.SimpleProducerConsumerBuffer;

public class SocketChannelHandler extends StateMachine {

	public static enum ChannelState {
		READING, WRITING, CLOSING, CLOSED;
	}

	private final static int TRANSFER_SIZE = 8096;
	
	private final SelectionKey key;
	private final SocketChannel channel;
	private final ByteBuffer transferBuffer;
	private final ProducerConsumerBuffer writeBuffer;
	private final ProducerConsumerBuffer readBuffer;

	private ChannelState channelState;
	
	public SocketChannelHandler(SelectionKey sk) {
		this.key = sk;
		this.channel = (SocketChannel) sk.channel();
		this.transferBuffer = ByteBuffer.allocateDirect(TRANSFER_SIZE);
		this.writeBuffer = new SimpleProducerConsumerBuffer();
		this.readBuffer = new SimpleProducerConsumerBuffer();
		
		this.channelState = ChannelState.WRITING;
		setReading();
	}

	public final void writeNow() throws IOException {
		
		if (channelState == ChannelState.CLOSED) {
			throw new IOException("Closed");
		}
		
		int toWrite = writeBuffer.getReadable();
		while (toWrite > 0) {
			transferBuffer.clear();
			int realToWrite = Math.min(TRANSFER_SIZE, toWrite);
			writeBuffer.produceBytes(transferBuffer, realToWrite);
			transferBuffer.flip();
			channel.write(transferBuffer);
			
			int remaining = transferBuffer.remaining();
			if (remaining > 0) {
				writeBuffer.revokeBytes(remaining);
				return;
			}
			
			toWrite -= realToWrite;
		}
		
		if (channelState == ChannelState.CLOSING) {
			disconnect();
		} else {
			setReading();
		}
	}

	private void setClosed() {
		key.interestOps(SelectionKey.OP_WRITE);
		this.channelState = ChannelState.CLOSED;
	}
	
	private void setClosing() {
		key.interestOps(SelectionKey.OP_WRITE);
		this.channelState = ChannelState.CLOSING;
	}
	
	private void setWriting() {
		if (channelState == ChannelState.READING) {
			key.interestOps(SelectionKey.OP_READ | SelectionKey.OP_WRITE);
		}
		this.channelState = ChannelState.WRITING;
	}
	
	private void setReading() {
		if (channelState == ChannelState.WRITING) {
			key.interestOps(SelectionKey.OP_READ);
		}
		this.channelState = ChannelState.READING;
	}

	public final void readNow() throws IOException {
		if (channelState != ChannelState.CLOSING) {
			while (true) {
				if (readBuffer.getWritable() < TRANSFER_SIZE) {
					break;
				}
				transferBuffer.clear();
				int read = channel.read(transferBuffer);
				if (read == -1) {
					throw new IOException("Closed");
				}
				transferBuffer.flip();
				readBuffer.consumeBytes(transferBuffer, read);
				if (read < TRANSFER_SIZE) {
					break;
				}
			}
			if (readBuffer.getReadable() > 0) {
				onBytesAvailable();
			}
		}
	}

	@Override
	public final void consumeByte(byte b) {
		if (channelState != ChannelState.CLOSING) {
			writeBuffer.consumeByte(b);
			setWriting();
		}
	}

	@Override
	public final void consumeInt(int i) {
		if (channelState != ChannelState.CLOSING) {
			writeBuffer.consumeInt(i);
			setWriting();
		}
	}

	@Override
	public final void consumeDouble(double d) {
		if (channelState != ChannelState.CLOSING) {
			writeBuffer.consumeDouble(d);
			setWriting();
		}
	}

	@Override
	public final void consumeBytes(byte[] buffer, int offset, int length) {
		if (channelState != ChannelState.CLOSING) {
			writeBuffer.consumeBytes(buffer, offset, length);
			setWriting();
		}
	}

	@Override
	public void consumeBytes(ByteBuffer bb, int length) {
		if (channelState != ChannelState.CLOSING) {
			writeBuffer.consumeBytes(bb, length);
			setWriting();
		}
	}

	@Override
	public byte produceByte() {
		return readBuffer.produceByte();
	}

	@Override
	public int produceInt() {
		return readBuffer.produceInt();
	}

	@Override
	public double produceDouble() {
		return readBuffer.produceDouble();
	}

	@Override
	public void produceBytes(ByteBuffer bb, int length) {
		readBuffer.produceBytes(bb, length);
	}

	@Override
	public void produceBytes(byte[] bs, int offset, int length) {
		readBuffer.produceBytes(bs, offset, length);
	}

	@Override
	public int getBytesAvailable() {
		return readBuffer.getReadable();
	}

	public void disconnectOnError(ConnectionListener listener) {
		listener.disconnected(this);
		key.cancel();
	}

	@Override
	public void disconnect() {
		setClosed();
	}

	@Override
	public void disconnectAfterAllWrites() {
		if (writeBuffer.getReadable() > 0) {
			setClosing();
		} else {
			setClosed();
		}
	}

}
